package cn.lzm.registry;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import cn.lzm.config.RegistryConfig;
import cn.lzm.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.DeleteResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @author lzm
 * @Date 2024/8/6
 * @description ETCD实现的注册中心
 */
public class EtcdRegistry implements Registry {

    private static final Logger log = LoggerFactory.getLogger(EtcdRegistry.class);

    private Client client;

    private KV kvClient;

    /**
     * 根节点
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";

    /**
     * 本机已注册的节点key列表，其实就是不同的controller类，比如：UserController, OrderController，
     */
    private static final Set<String> LOCAL_REGISTER_NODE_KEYS = new ConcurrentHashSet<>();

    /**
     * 注册中心缓存
     */
    private static final RegistryServiceCache REGISTRY_SERVICE_CACHE = new RegistryServiceCache();

    /**
     * 正在监听的key集合
     */
    private static final Set<String> WATCHING_KEYS = new ConcurrentSkipListSet<>();

    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder().endpoints(registryConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo metaInfo) throws Exception {
        Lease leaseClient = client.getLeaseClient();
        
        // 设置超时时间30s，防止请求总是转发到一个实际上已经停止工作的服务器
        long leaseId = leaseClient.grant(30).get().getID();

        String registerKey = ETCD_ROOT_PATH + metaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence val = ByteSequence.from(JSONUtil.toJsonStr(metaInfo), StandardCharsets.UTF_8);

        LOCAL_REGISTER_NODE_KEYS.add(registerKey);
        // 设置服务节点信息的过期时间
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, val, putOption).get();
    }

    @Override
    public void unregister(ServiceMetaInfo metaInfo) throws Exception {
        String registerKey = ETCD_ROOT_PATH + metaInfo.getServiceNodeKey();
        DeleteResponse resp = kvClient.delete(ByteSequence.from(registerKey, StandardCharsets.UTF_8)).get();
        log.info("服务注销结果：{}", JSONUtil.toJsonStr(resp));
    }

    @Override
    public List<ServiceMetaInfo> discoverService(String service) {
        List<ServiceMetaInfo> cache = REGISTRY_SERVICE_CACHE.readCache(service);
        if (CollectionUtil.isNotEmpty(cache)) {
            return cache;
        }
        String searchPrefix = ETCD_ROOT_PATH + service + "/";
        // 使用前缀匹配，也就是左模糊查询
        GetOption getOption = GetOption.builder().isPrefix(true).build();
        List<KeyValue> kvs;
        try {
            kvs = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), getOption)
                    .get()
                    .getKvs();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        if (CollectionUtil.isEmpty(kvs)) {
            throw new RuntimeException(String.format("%s的服务不存在", service));
        }
        try {
            List<ServiceMetaInfo> info = kvs.stream().map(
                    kv -> {
                        watch(kv.getKey().toString(StandardCharsets.UTF_8));
                        String value = kv.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    }
            ).collect(Collectors.toList());
            REGISTRY_SERVICE_CACHE.writeCache(service, info);
            return info;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败");
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");
        for (String key : LOCAL_REGISTER_NODE_KEYS) {
            try {
                // 注意：这里在最后面一定要加上get方法才能实现同步删除key的效果
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
                System.out.printf("节点%s已下线\n", key);
            } catch (Exception e) {
                throw new RuntimeException(key + "节点下线失败", e);
            }
        }
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }

    /**
     * 定期刷新每个节点的过期时间
     */
    @Override
    public void heartBeat() {
        CronUtil.schedule("*/10 * * * * ?", new Task() {
            @Override
            public void execute() {
                for (String key : LOCAL_REGISTER_NODE_KEYS) {
                    try {
                        List<KeyValue> kvs = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get().getKvs();
                        if (CollectionUtil.isEmpty(kvs)) {
                            continue;
                        }
                        KeyValue kv = kvs.get(0);
                        String value = kv.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo metaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(metaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + "续签失败", e);
                    }
                }
            }
        });
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        boolean newWatch = WATCHING_KEYS.add(serviceNodeKey);
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    if (Objects.requireNonNull(event.getEventType()) == WatchEvent.EventType.DELETE) {
                        REGISTRY_SERVICE_CACHE.clearCache(serviceNodeKey);
                    }
                }
            });
        }
    }
}
